package com.atguigu.gmall.app.dwd.db;

import com.atguigu.gmall.util.KafkaUtil;
import com.atguigu.gmall.util.MySQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @author yhm
 * @create 2022-09-14 10:17
 */
public class DwdTradeOrderRefund {
    public static void main(String[] args) {
        // TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5L));

        // TODO 2 设置状态后端
        /*
        env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE );
        env.getCheckpointConfig().setCheckpointTimeout( 3 * 60 * 1000L );
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        System.setProperty("HADOOP_USER_NAME", "atguigu");
         */

        // TODO 3 读取topic_db数据
        String topicName = "topic_db";
        String groupId = "dwd_trade_order_refund";
        tableEnv.executeSql("CREATE TABLE topic_db (\n" +
                "  `database` STRING,\n" +
                "  `table` STRING,\n" +
                "  `type` STRING,\n" +
                "  `ts` bigint,\n" +
                "  `data` MAP<STRING,STRING>,\n" +
                "  `old` MAP<STRING,STRING>,\n" +
                "  `pt` as proctime()\n" +
                ")" + KafkaUtil.getKafkaDDL(topicName,groupId));

        // TODO 4 过滤退单表数据
        Table refundInfo = tableEnv.sqlQuery("select \n" +
                "  data['order_id'] order_id,\n" +
                "  data['sku_id'] sku_id,\n" +
                "  data['refund_type'] refund_type,\n" +
                "  data['refund_num'] refund_num,\n" +
                "  data['refund_amount'] refund_amount,\n" +
                "  data['refund_reason_type'] refund_reason_type,\n" +
                "  data['refund_reason_txt'] refund_reason_txt,\n" +
                "  data['refund_status'] refund_status,\n" +
                "  `ts`,\n" +
                "  `pt`\n" +
                "from topic_db\n" +
                "where `table`='order_refund_info'\n" +
                "and `type`='insert'");

        tableEnv.createTemporaryView("refund_info",refundInfo);

        // TODO 5 过滤订单表数据状态为退款中
        Table orderInfo = tableEnv.sqlQuery("select \n" +
                "  data['id'] id,\n" +
                "  data['consignee'] consignee,\n" +
                "  data['consignee_tel'] consignee_tel,\n" +
                "  data['total_amount'] total_amount,\n" +
                "  data['user_id'] user_id,\n" +
                "  data['payment_way'] payment_way,\n" +
                "  data['delivery_address'] delivery_address,\n" +
                "  data['order_comment'] order_comment,\n" +
                "  data['out_trade_no'] out_trade_no,\n" +
                "  data['trade_body'] trade_body,\n" +
                "  data['create_time'] create_time,\n" +
                "  data['operate_time'] operate_time,\n" +
                "  data['expire_time'] expire_time,\n" +
                "  data['process_status'] process_status,\n" +
                "  data['tracking_no'] tracking_no,\n" +
                "  data['parent_order_id'] parent_order_id,\n" +
                "  data['province_id'] province_id,\n" +
                "  data['activity_reduce_amount'] activity_reduce_amount,\n" +
                "  data['coupon_reduce_amount'] coupon_reduce_amount,\n" +
                "  data['original_total_amount'] original_total_amount,\n" +
                "  data['feight_fee'] feight_fee,\n" +
                "  data['feight_fee_reduce'] feight_fee_reduce,\n" +
                "  data['refundable_time'] refundable_time\n" +
                "from topic_db\n" +
                "where `table`='order_info'\n" +
                "and data['order_status']='1005' and `type`='update'");
        tableEnv.createTemporaryView("order_info",orderInfo);

        // TODO 6 读取字典表数据
        tableEnv.executeSql(MySQLUtil.getBaseDicDDL());

        // TODO 7 关联3张表
        Table refundOrderTable = tableEnv.sqlQuery("select \n" +
                "  ri.order_id ,\n" +
                "  ri.sku_id ,\n" +
                "  bd.dic_name refund_type ,\n" +
                "  ri.refund_num ,\n" +
                "  ri.refund_amount ,\n" +
                "  ri.refund_reason_type ,\n" +
                "  ri.refund_reason_txt ,\n" +
                "  ri.refund_status ,\n" +
                "  oi.consignee ,\n" +
                "  oi.consignee_tel ,\n" +
                "  oi.total_amount ,\n" +
                "  oi.user_id ,\n" +
                "  oi.payment_way ,\n" +
                "  oi.delivery_address ,\n" +
                "  oi.order_comment ,\n" +
                "  oi.out_trade_no ,\n" +
                "  oi.trade_body ,\n" +
                "  oi.create_time ,\n" +
                "  oi.operate_time ,\n" +
                "  oi.expire_time ,\n" +
                "  oi.process_status ,\n" +
                "  oi.tracking_no ,\n" +
                "  oi.parent_order_id ,\n" +
                "  oi.province_id ,\n" +
                "  oi.activity_reduce_amount ,\n" +
                "  oi.coupon_reduce_amount ,\n" +
                "  oi.original_total_amount ,\n" +
                "  oi.feight_fee ,\n" +
                "  oi.feight_fee_reduce ,\n" +
                "  oi.refundable_time ,\n" +
                "  ts ,\n" +
                "  pt \n" +
                "from refund_info ri\n" +
                "join order_info oi\n" +
                "on oi.id=ri.order_id\n" +
                "join base_dic FOR SYSTEM_TIME AS OF ri.pt as bd\n" +
                "on bd.dic_code=ri.refund_type");
        tableEnv.createTemporaryView("refund_order",refundOrderTable);

        // TODO 8 写出到kafka
        tableEnv.executeSql("create table refund_order_result(\n" +
                "  order_id STRING,\n" +
                "  sku_id STRING,\n" +
                "  refund_type STRING,\n" +
                "  refund_num STRING,\n" +
                "  refund_amount STRING,\n" +
                "  refund_reason_type STRING,\n" +
                "  refund_reason_txt STRING,\n" +
                "  refund_status STRING,\n" +
                "  consignee STRING,\n" +
                "  consignee_tel STRING,\n" +
                "  total_amount STRING,\n" +
                "  user_id STRING,\n" +
                "  payment_way STRING,\n" +
                "  delivery_address STRING,\n" +
                "  order_comment STRING,\n" +
                "  out_trade_no STRING,\n" +
                "  trade_body STRING,\n" +
                "  create_time STRING,\n" +
                "  operate_time STRING,\n" +
                "  expire_time STRING,\n" +
                "  process_status STRING,\n" +
                "  tracking_no STRING,\n" +
                "  parent_order_id STRING,\n" +
                "  province_id STRING,\n" +
                "  activity_reduce_amount STRING,\n" +
                "  coupon_reduce_amount STRING,\n" +
                "  original_total_amount STRING,\n" +
                "  feight_fee STRING,\n" +
                "  feight_fee_reduce STRING,\n" +
                "  refundable_time STRING,\n" +
                "  ts BIGINT,\n" +
                "  pt TIMESTAMP_LTZ(3)\n" +
                ")" + KafkaUtil.getKafkaSinkDDL("dwd_trade_order_refund"));

        tableEnv.executeSql("insert into refund_order_result select * from refund_order");

    }
}
